package cn.edu.cuit.elena.web.service;

import java.io.IOException;
import java.io.PrintWriter;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import cn.edu.cuit.elena.common.Configuration;
import cn.edu.cuit.elena.common.SackConstants;
import cn.edu.cuit.elena.db.DataSource;
import cn.edu.cuit.elena.db.DataSourceFactory;
import cn.edu.cuit.elena.db.Schema;
import cn.edu.cuit.elena.db.Schema.Table;
import cn.edu.cuit.elena.trace.Tracer;
import cn.edu.cuit.elena.transaction.controller.Hive;
import cn.edu.cuit.elena.transaction.controller.Sqoop;
import cn.edu.cuit.elena.transaction.system.UserContext;
import cn.edu.cuit.elena.web.System;
import cn.edu.cuit.elena.web.service.ServiceServlet.ServiceConstants;
import cn.edu.cuit.elena.web.tool.JsonHelper;

public class DataMigrationServlet
    extends HttpServlet
{

    static final String LOAD_DATASOURCE = "datasource";
    static final String DO_MIGRATE = "migrate";
    static final String COLUMN = "column";
    static final String TABLE = "table";
    static final String TEST = "test";
    static final String MIGRATING = "migrating";

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    /**
     * There is a List which contains that <strong>datasource properties map which has been taken it
     * into migrating</strong> by each userContext.
     * 
     * <pre>
     * A datasouce
     * 
     * map.put( &quot;username&quot;, username );
     * map.put( &quot;password&quot;, password );
     * map.put( &quot;url&quot;, url );
     * map.put( &quot;driver&quot;, driver );
     * *****************************************
     * 
     * A cacheList for userContext
     * 
     * cacheList[
     *              map_1,
     *              map_2,
     *              map_3,
     *              ... ...
     *          ]
     * *****************************************
     * 
     * Cache
     * 
     * Map[
     *          cacheList 1 for userContext.getIdenfifier 1,
     *          cacheList 1 for userContext.getIdenfifier 2,
     *          cacheList 1 for userContext.getIdenfifier 3,
     *          ... ...
     *      ]
     * </pre>
     * 
     * Using <code>cache#get(userContext.getIdentifier)</code> could get this cached list.
     * 
     * */
    volatile static Map<String, List<Map<String, String>>> cache;

    {
        cache = new HashMap<String, List<Map<String, String>>>();
    }

    /**
     * Destruction of the servlet. <br>
     */
    public void destroy()
    {
        super.destroy(); // Just puts "destroy" string in log
        // Put your code here
    }

    /**
     * The doGet method of the servlet. <br>
     * 
     * This method is called when a form has its tag value method equals to get.
     * 
     * @param request
     *            the request send by the client to the server
     * @param response
     *            the response send by the server to the client
     * @throws ServletException
     *             if an error occurred
     * @throws IOException
     *             if an error occurred
     */
    public void doGet( HttpServletRequest request, HttpServletResponse response ) throws ServletException, IOException
    {

        doPost( request, response );
    }

    /**
     * The doPost method of the servlet. <br>
     * 
     * This method is called when a form has its tag value method equals to post.
     * 
     * @param request
     *            the request send by the client to the server
     * @param response
     *            the response send by the server to the client
     * @throws ServletException
     *             if an error occurred
     * @throws IOException
     *             if an error occurred
     */
    public void doPost( HttpServletRequest request, HttpServletResponse response ) throws ServletException, IOException
    {

        response.setContentType( "text/html" );
        PrintWriter out = response.getWriter();
        String para = request.getParameter( "kind" );
        UserContext userContext = ( (UserContext) request.getSession().getAttribute(
            ServiceConstants.USERCONTEXT.mark() ) );

        String username = request.getParameter( "username" );
        String password = request.getParameter( "password" );
        String url = request.getParameter( "url" );
        String driver = request.getParameter( "driver" );

        Map<String, String> map = new HashMap<String, String>();
        map.put( "username", username );
        map.put( "password", password );
        map.put( "url", url );
        map.put( "driver", driver );

        if( LOAD_DATASOURCE.equals( para ) )
        {
            String json = JsonHelper.getNewInstance().toDataSourcesJson( userContext.getCache().getCachesDataSources() );

            Tracer.debugTrace( DataMigrationServlet.class, "DataMigrationServlet datasource json : " + json, null );
            out.write( json );
            out.flush();
            out.close();
        }
        else if( DO_MIGRATE.equals( para ) )
        {
            Sqoop sqoop = Sqoop.get( userContext, System.getSystemTaskMonitor() );
            String dbId = url + username + password + driver;
            String tableName = request.getParameter( "tablename" );

            if( tableName == null || tableName.equals( "null" ) )
            {
                out.write( SackConstants.FAILURE );
                out.flush();
                out.close();

                return;
            }

            //TO be using.
            String taskId = dbId + tableName;

            List<Map<String, String>> cacheList = cache.get( userContext.getIdentifier() );

            if( cacheList == null )
            {
                cacheList = new LinkedList<Map<String, String>>();
            }

            for( Map<String, String> map2 : cacheList )
            {
                //If there already is a datasource in cache, then don't do migration.
                if( map2.get( "username" ).equals( username ) && map2.get( "password" ).equals( password )
                    && map2.get( "url" ).equals( url ) && map2.get( "driver" ).equals( driver )
                    && map2.get( "tablename" ).equals( tableName ) )
                {
                    out.write( String.valueOf( SackConstants.DUPLICATED ) );
                    out.close();
                    out.flush();

                    return;
                }
            }

            //            //Task id = dbId + tableName
            //            /**
            //             * Here! do import.
            //             * */
            sqoop.doImport( dbId, tableName );

            DataSource dataSource = null;
            for( DataSource dataSource0 : userContext.getCache().getCachesDataSources() )
            {
                if( dataSource0.toString().equals( url + username + password + driver ) )
                {
                    dataSource = dataSource0;
                }
            }

            if( dataSource == null )
            {
                out.close();
                return;
            }

            //            /**
            //             * Here! create hive table and associate with import data.
            //             * */
            //            Hive hive = Hive.get( userContext, dataSource );
            //            try
            //            {
            //                hive.doCreate( tableName );
            //                hive.loadDataIntoHiveTable( tableName );
            //            }
            //            catch( SQLException e )
            //            {
            //                Tracer.fatalTrace( DataMigrationServlet.class,
            //                    "Can create hive table and load data due to : " + e.getMessage(), e );
            //
            //                out.close();
            //                return;
            //            }

            Map<String, String> cacheMap = new HashMap<String, String>();
            cacheMap.put( "taskid", dbId + tableName );
            cacheMap.put( "username", username );
            cacheMap.put( "password", password );
            cacheMap.put( "url", url );
            cacheMap.put( "driver", driver );
            cacheMap.put( "tablename", tableName );
            cacheMap.put( "progress",
                String.valueOf( System.getSystemTaskMonitor().getTaskById( dbId + tableName ).getProgression() ) );

            cacheList.add( cacheMap );

            out.write( SackConstants.SUCCESS );
            out.flush();
            out.close();

        }
        else if( "delete".equals( para ) )
        {
            for( DataSource d : userContext.getCache().getCachesDataSources() )
            {
                if( d.toString().equalsIgnoreCase( url + username + password + driver ) )
                {
                    userContext.getCache().getCachesDataSources().remove( d );
                }
            }
        }
        else if( TABLE.equals( para ) )
        {
            List<DataSource> list = userContext.getCache().getCachesDataSources();

            for( DataSource dataSource : list )
            {
                if( dataSource.toString().equals( url + username + password + driver ) )
                {
                    try
                    {
                        Schema schema = dataSource.loadSchema();

                        String json = JsonHelper.getNewInstance().toTablesJson( schema.getTables() );

                        out.write( json );
                        out.flush();
                        out.close();

                        Tracer.debugTrace( DataMigrationServlet.class, "DataMigrationServlet Tables json : " + json,
                            null );
                    }
                    catch( Exception e )
                    {
                        Tracer.fatalTrace( DataMigrationServlet.class,
                            "DataMigrationServlet : can't load schema for DataSource : " + url + username + password
                                + driver, e );

                        out.write( SackConstants.FAILURE );
                        out.flush();
                        out.close();
                    }
                }
            }
        }
        //I don't guarantee that, each DataSource Object can be created successfully,
        //so , if creation failed, clicking on "Table Details" button will show nothing.
        //But i do guarantee that the subsequent operating which happened after clicking 
        //on "Test Data Source" button, and the "state" field shows "Ready" works correctly.
        else if( COLUMN.equals( para ) )
        {
            List<DataSource> list = userContext.getCache().getCachesDataSources();

            String tableName = request.getParameter( "tablename" );
            String schemaName = request.getParameter( "schemaname" );

            for( DataSource dataSource : list )
            {
                if( dataSource.toString().equals( url + username + password + driver )
                    && dataSource.getName().equals( schemaName ) )
                {
                    try
                    {
                        Schema schema = dataSource.loadSchema();
                        Table table = schema.getTables().get( tableName );

                        String json = JsonHelper.getNewInstance().toColumnJson( table );

                        out.write( json );
                        out.flush();
                        out.close();

                        Tracer.debugTrace( DataMigrationServlet.class, "DataMigrationServlet Table json : " + json,
                            null );
                    }
                    catch( SQLException e )
                    {
                        Tracer.fatalTrace( DataMigrationServlet.class,
                            "DataMigrationServlet : can't load schema for DataSource : " + url + username + password
                                + driver, e );

                        out.write( SackConstants.FAILURE );
                        out.flush();
                        out.close();
                    }
                }
            }
        }
        else if( TEST.equals( para ) )
        {
            try
            {
                DataSourceFactory.createDataSourceByConfiguration( Configuration.create( SackConstants.URL + "=" + url,
                    SackConstants.USER_NAME + "=" + username, SackConstants.PASS_WORD + "=" + password,
                    SackConstants.DRIVER + "=" + driver ) );

                out.write( SackConstants.SUCCESS );
                out.flush();
                out.close();
            }
            catch( Exception e )
            {
                out.write( SackConstants.FAILURE );
                out.flush();
                out.close();
            }
        }
        else if( MIGRATING.equals( para ) )
        {
            List<Map<String, String>> cacheList = cache.get( userContext.getIdentifier() );

            if( cacheList == null )
            {
                cache.put( userContext.getIdentifier(), new LinkedList<Map<String, String>>() );
            }

            for( Map<String, String> map0 : cacheList )
            {
                String progress;
                if( System.getSystemTaskMonitor().getTaskById( map0.get( "taskid" ) ) == null )
                {
                    progress = System.getSystemTaskMonitor().getRetiredTaskById( map0.get( "taskid" ) )
                        .getProgression()
                        + "";
                }
                else
                {

                    progress = System.getSystemTaskMonitor().getTaskById( map0.get( "taskid" ) ).getProgression() + "";
                }

                if( !map0.get( "progress" ).equals( progress ) )
                {
                    map0.put( "progress", progress );
                }
            }

            String json = JsonHelper.getNewInstance().toJson( cacheList );

            Tracer.debugTrace( DataMigrationServlet.class, "DataMigrationServlet migrating json : " + json, null );
            out.write( json );
            out.close();
            out.flush();
        }
    }

    /**
     * Initialization of the servlet. <br>
     * 
     * @throws ServletException
     *             if an error occurs
     */
    public void init() throws ServletException
    {
        // Put your code here
    }

}
